package com.kamistoat.meimeistore.seckill.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.kamistoat.common.To.LoginSuccessTo;
import com.kamistoat.common.To.RabbitMQTo.SecKillOrderTo;
import com.kamistoat.common.To.SecKillTo.SecKillProductCacheTo;
import com.kamistoat.common.To.SecKillTo.SecKillProductTo;
import com.kamistoat.common.To.SecKillTo.SecKillSessionTo;
import com.kamistoat.common.To.SecKillTo.SecKillSkuInfoTo;
import com.kamistoat.common.constant.SecKillConstant;
import com.kamistoat.common.runTimeFunc.RunTimeLog;
import com.kamistoat.common.utils.R;
import com.kamistoat.meimeistore.seckill.feign.CouponFeignService;
import com.kamistoat.meimeistore.seckill.feign.ProductFeignService;
import com.kamistoat.meimeistore.seckill.interceptor.LoginInterceptor;
import com.kamistoat.meimeistore.seckill.service.SecKillService;
import com.kamistoat.meimeistore.seckill.vo.SecKillProductCacheVo;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.*;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Service
public class SecKillServiceImpl implements SecKillService {

    @Autowired
    CouponFeignService couponFeignService;
    @Autowired
    ProductFeignService productFeignService;
    @Autowired
    StringRedisTemplate stringRedisTemplate;
    @Autowired
    RedissonClient redissonClient;
    @Autowired
    RabbitTemplate rabbitTemplate;

    /**
     * 上架未来三天的秒杀商品
     */
    @Override
    public void upload3DaySecKillSku() {
        // 1远程查询参与秒杀的活动和商品。需要编写远程接口去查coupon系统中未来三天的秒杀活动
        List<SecKillSessionTo> sessionToList = couponFeignService.get3DaySession();
        // 2将活动和商品缓存到redis中
        if (sessionToList == null || sessionToList.size() <= 0) {
            return;
        } else {
            redisSceSession(sessionToList);
            redisSceProduct(sessionToList);
            return;
        }
    }

    /**
     * 删除redis中已经缓存的原秒杀
     * 并重新检查相同id是否是3天以内的秒杀，如果是，立刻上架
     */
    @Override
    public void delete_ReUp_RedisSessions(List<String> redisSessionKeyList) {
        for (String sessionKey : redisSessionKeyList) {
            // 如果redis中含有这个key，说明已经缓存过了，那就删除之
            if (stringRedisTemplate.hasKey(sessionKey)) {
                extracted_DeleteRedisSession(sessionKey);
            }
        }
        this.upload3DaySecKillSku();
    }

    /**
     * 删除已经过期一天的活动+商品+信号量+商品_用户记录
     */
    @Override
    public void deleteOut1DaySecKill() {
        // 1 查出所有活动
        long curtime = new Date().getTime();
        // redis毕竟不是ES，所以只能是先把所有的秒杀场次都查出来，然后筛选出包含当前时间的场次
        // 用的是正则表达式，查询出所有与seckill:sessions:*匹配的redis键名。注意只是键，不是键值对
        Set<String> sessionKeySet = stringRedisTemplate.keys(SecKillConstant.SEC_SESSION_PREFIX + "*");

        // 对每一场活动，判断活动的结束时间是否比 今天-1天 还早
        for (String sessionKey : sessionKeySet) {
            // seckill:sessions:1638961200000_1638964800000
            String sessionTime = sessionKey.replace(SecKillConstant.SEC_SESSION_PREFIX, "");
            Long sessionEndtime = Long.parseLong(sessionTime.split("_")[1]);
            if (curtime - 1000 * 60 * 60 * 24 >= sessionEndtime) {
                // 2 如果找到了已经过期一天的活动，查这场的商品信息
                extracted_DeleteRedisSession(sessionKey);
            }
        }
    }

    /**
     * 抽取的方法。从redis中删除秒杀场次以及关联的全部数据
     *
     * @param sessionKey
     */
    private void extracted_DeleteRedisSession(String sessionKey) {
        RunTimeLog.printRunTimeLog("删除秒杀 " + sessionKey + " 在Redis中的缓存");
        // 由于秒杀活动键值对的值是一个list<场次-商品关联id>，要获取这个List稍微有些麻烦。
        // 用opsForList().range(键名，起始下标，终点下标)获取List的值。因为要全部取出来，直接以Long的最大范围做值
        List<String> sessionProIdList = stringRedisTemplate.opsForList().range(sessionKey, Long.MIN_VALUE, Long.MAX_VALUE);
        // 得到当前场次的所有场次-商品关联id，然后根据这个id接着查商品SecKillProductCacheVo信息
        // 这个绑定操作的<>内分别代表<这个哈希表在redis中的键数据类型，哈希表内部-键值对-键数据类型，哈希表内部-键值对-值数据类型>
        // 显然全都是String，因为我们早就都统一使用JSON做序列化机智了
        BoundHashOperations<String, String, String> sessionProHash =
                stringRedisTemplate.boundHashOps(SecKillConstant.SEC_PRODUCT_PREFIX);
        // 现在绑定的这个哈希表中的键值对，key是场次-商品关联主键Id，value是一个JSON格式的Vo类
        // redis哈希表有一个快捷操作，输入一个列表，返回键值对中key位于列表中的value列表。返回一个List<Object>，显然这些Object全是JSONString，直接写就行了
        assert sessionProIdList != null;
        List<String> secProductToJSONList = sessionProHash.multiGet(sessionProIdList);
        if (secProductToJSONList != null && secProductToJSONList.size() > 0) {
            // 得到的这个列表中就是当前场次对应的所有 场次-商品关联Vo
            // 根据catchVo中的随机码去删除信号量锁
            for (String json : secProductToJSONList) {
                SecKillProductCacheVo productCacheVo = JSON.parseObject(json.toString(), SecKillProductCacheVo.class);
                stringRedisTemplate.delete(SecKillConstant.SEC_KILL_SEMAPHORE + productCacheVo.getRandomCode());
            }
        }
        // 删完信号量锁，再从哈希表中把对应的商品catchVo删除。同时一并删除商品_用户记录
        for (String secproId : sessionProIdList) {
            sessionProHash.delete(secproId);  // 因为sessionProIdList中保存是哈希表中的key，直接在哈希表中根据key删除value即可
            // 把这个场次商品关联Id对应的所有的商品_用户购买记录信息也删除
            Set<String> proUsers = stringRedisTemplate.keys(SecKillConstant.SEC_PRO_USER + secproId + "*");
            assert proUsers != null;
            stringRedisTemplate.delete(proUsers);
        }
        // 最后再删秒杀活动
        stringRedisTemplate.delete(sessionKey);
    }

    /**
     * 获取当前时间段秒杀活动的商品信息
     * 如果不存在，就返回下一场秒杀
     *
     * @return
     */
    @Override
    public List<SecKillProductCacheTo> getCurrentSecKillSession() {
        List<List<SecKillProductCacheTo>> futureSecKillSessionList = getFutureSecKillSessionList();
        if (!futureSecKillSessionList.isEmpty() && futureSecKillSessionList.size() > 0) {
            return futureSecKillSessionList.get(0);
        } else {
            return null;
        }
    }

    /**
     * 获取未来所有的秒杀场次的商品信息
     * 外层List的每一个元素是一场秒杀
     * 每层List的每一个元素是这场秒杀的商品信息
     *
     * @return
     */
    @Override
    public List<List<SecKillProductCacheTo>> getFutureSecKillSessionList() {
        // 制作优先队列
        PriorityQueue<List<SecKillProductCacheTo>> futureSecKillSessionQueue =
                new PriorityQueue<List<SecKillProductCacheTo>>(Comparator.comparing(l -> l.get(0).getStartTime()));

        // 直接到redis中，去查未来的所有活动（根据键来查，因为键是活动的起始结束时间）
        // 再根据活动的值（场次-商品关联id）去查关联id为键下的商品信息

        // 1 查出所有活动
        long curtime = new Date().getTime();
        // redis毕竟不是ES，所以只能是先把所有的秒杀场次都查出来，然后筛选出未来时间的场次
        // 用的是正则表达式，查询出所有与seckill:sessions:*匹配的redis键名。注意只是键，不是键值对
        Set<String> sessionKeySet = stringRedisTemplate.keys(SecKillConstant.SEC_SESSION_PREFIX + "*");

        // 对每一场活动，判断活动的结束时间是否比现在晚
        for (String sessionKey : sessionKeySet) {
            // seckill:sessions:1638961200000_1638964800000
            String sessionTime = sessionKey.replace(SecKillConstant.SEC_SESSION_PREFIX, "");
            Long sessionStartTime = Long.parseLong(sessionTime.split("_")[0]);
            Long sessionEndtime = Long.parseLong(sessionTime.split("_")[1]);

            // 只要结束时间比现在晚，那就是一场未来的秒杀活动
            if (curtime <= sessionEndtime) {
                // 2 如果找到了一个未来时间的秒杀活动，查这场的商品信息
                // 由于秒杀活动键值对的值是一个list<场次-商品关联id>，要获取这个List稍微有些麻烦。
                // 用opsForList().range(键名，起始下标，终点下标)获取List的值。因为要全部取出来，直接以Long的最大范围做值
                List<String> sessionProIdList = stringRedisTemplate.opsForList().range(sessionKey, Long.MIN_VALUE, Long.MAX_VALUE);
                // 检验。如果这场秒杀活动没有任何秒杀商品，相当于没有这场活动
                assert sessionProIdList != null;
                // 得到当前场次的所有场次-商品关联id，然后根据这个id接着查商品SecKillProductCacheVo信息
                // 这个绑定操作的<>内分别代表<这个哈希表在redis中的键数据类型，哈希表内部键值对键数据类型，哈新表内部键值对值数据类型>
                // 显然全都是String，因为我们早就都统一使用JSON做序列化机智了
                BoundHashOperations<String, String, String> sessionProHash =
                        stringRedisTemplate.boundHashOps(SecKillConstant.SEC_PRODUCT_PREFIX);
                // redis哈希表有一个快捷操作，可以对一个键列表进行批量查询。返回一个List<Object>，显然这些Object全是JSONString，直接写就行了
                List<String> secProductToJSONList = sessionProHash.multiGet(sessionProIdList);
                // 得到的 secProductToJSONList 中的每一个元素都是当前秒杀场次 中的 秒杀商品。遍历处理之
                // 所以还要把JSON转换成To，因为要进行类间传输。需要建立一个和SecKillProductCacheVo完全相同的To即可。
                if (secProductToJSONList != null && secProductToJSONList.size() > 0) {
                    List<SecKillProductCacheTo> tempSecKill = new ArrayList<>();
                    for (String secProductToJSON : secProductToJSONList) {
                        SecKillProductCacheVo productCacheVo = JSON.parseObject(secProductToJSON.toString(), SecKillProductCacheVo.class);
                        // 直接把Redis中保存的Vo复制到To上即可，因为两个类内容是完全一样的所以直接拷贝
                        SecKillProductCacheTo productCacheTo = new SecKillProductCacheTo();
                        BeanUtils.copyProperties(productCacheVo, productCacheTo);
                        // 这里如何处理随机码有一层逻辑在。秒杀开始后，得返回随机码。秒杀没开始，不返回随机码。
                        // 很简单，判断一下时间即可。如果秒杀尚未开始，就将To的randomCode置为0
                        if (curtime < sessionStartTime) {
                            productCacheTo.setRandomCode("");
                        }
                        // 加入当前秒杀场的列表
                        tempSecKill.add(productCacheTo);
                    }
                    // 当前场所有商品处理完后，将当前秒杀场列表 加入 优先队列。按照开始时间早晚排序
                    futureSecKillSessionQueue.offer(tempSecKill);
                }
            }
        }

        List<List<SecKillProductCacheTo>> futureSecKillSessionList = new ArrayList<>();
        while (!futureSecKillSessionQueue.isEmpty()) {
            futureSecKillSessionList.add(futureSecKillSessionQueue.poll());
        }
        return futureSecKillSessionList;
    }


    /**
     * 商品详情页，远程该商品未来三天是否有秒杀活动
     *
     * @param skuId skuId
     * @return 类间传输To
     */
    @Override
    public SecKillProductCacheTo getSkuSecKill(Long skuId) {
        long curtime = new Date().getTime();
        // 直接操作哈希表就行了
        BoundHashOperations<String, String, String> sessionProHash =
                stringRedisTemplate.boundHashOps(SecKillConstant.SEC_PRODUCT_PREFIX);
        // 获取哈希表中的所有key集合
        Set<String> secProKeys = sessionProHash.keys();
        // 最晚开始时间三天后
        Long startTime = curtime + 3 * 24 * 60 * 60 * 1000;
        for (String secProKey : secProKeys) {
            String cacheVoJSON = sessionProHash.get(secProKey);
            SecKillProductCacheVo cacheVo = JSON.parseObject(cacheVoJSON, SecKillProductCacheVo.class);
            if (cacheVo.getEndTime() > curtime &&
                    cacheVo.getSecKillSkuInfoTo().getSkuId().equals(skuId) &&
                    cacheVo.getStartTime() < startTime) {
                // 找出还未结束且离现在最近的该商品活动。
                SecKillProductCacheTo cacheTo = new SecKillProductCacheTo();
                BeanUtils.copyProperties(cacheVo, cacheTo);
                // 处理随机码--活动还未开始，不返回随机码，正在进行，返回随机码
                if (!(cacheTo.getStartTime() <= curtime && cacheTo.getEndTime() >= curtime)) {
                    cacheTo.setRandomCode("");
                }
                return cacheTo;
            }
        }
        return null;
    }

    /**
     * 缓存秒杀场次
     */
    private void redisSceSession(List<SecKillSessionTo> sessionToList) {
        for (SecKillSessionTo sessionTo : sessionToList) {
            // 将起始时间_结束时间作为 键
            // Date类型的数据可以使用.getTime()将数据转换成距离1970年毫秒数值的Long类型，方便比较
            Long startTime = sessionTo.getStartTime().getTime();
            Long endTime = sessionTo.getEndTime().getTime();
            // 制作当前秒杀场次的 redis 键
            String key = SecKillConstant.SEC_SESSION_PREFIX + startTime + "_" + endTime;
            Boolean hasKey = stringRedisTemplate.hasKey(key);
            // 先根据key判断是否已经上架过该活动，没有上架过才执行上架
            // 或者无论是否上架都执行上架，防止出现上架后又更改了的秒杀活动，因系统故障没有正确上架。
            if (!hasKey && sessionTo.getProductToList() != null && sessionTo.getProductToList().size() > 0) {
                // 而这个键的值是一个<String>列表，列表的每一个元素是当前这一场秒杀活动的某个 场次-商品关联信息的主键Id
                List<String> skuIdList = sessionTo.getProductToList().stream().map(secKillProductTo -> {
                    return secKillProductTo.getId().toString();
                }).collect(Collectors.toList());
                stringRedisTemplate.opsForList().leftPushAll(key, skuIdList);
                RunTimeLog.printRunTimeLog("Redis缓存秒杀场次: " + sessionTo.getName());
                // stringRedisTemplate.expire(key, 30L, TimeUnit.SECONDS);
            }
        }
    }

    /**
     * 缓存商品
     *
     * @param sessionToList
     */
    private void redisSceProduct(List<SecKillSessionTo> sessionToList) {
        // 由于所有场次的所有商品信息都保存在一个哈希表中，因此每次开始直接绑定这个哈希表就行了
        BoundHashOperations<String, Object, Object> productHash =
                stringRedisTemplate.boundHashOps(SecKillConstant.SEC_PRODUCT_PREFIX);
        for (SecKillSessionTo sessionTo : sessionToList) {
            if (sessionTo.getProductToList() != null && sessionTo.getProductToList().size() > 0) {
                // 对于未来三天的每一场秒杀场次，每个场次内可能有很多件秒杀商品，通过属性 productToList展示。遍历 productToList保存秒杀商品信息
                for (SecKillProductTo productTo : sessionTo.getProductToList()) {
                    // 对当前场次的每一个场次-商品关联，获取该商品的详细信息后，将关联+详情信息一起保存
                    // 当然在保存之前，必须先确认一下当前场次的这个场次-商品关联是否已经上架了，如果已经上架，关联信息+信号量全部跳过
                    Boolean hasKey = productHash.hasKey(productTo.getId().toString());
                    if (!hasKey) {
                        SecKillProductCacheVo cacheVo = new SecKillProductCacheVo();
                        SecKillSkuInfoTo skuInfoTo = productFeignService.getSecKillSkuTo(productTo.getSkuId());
                        cacheVo.setSecKillProductTo(productTo);  // 保存场次-商品关联信息，其中有商品在秒杀时的信息
                        cacheVo.setSecKillSkuInfoTo(skuInfoTo);  // 保存商品原始信息
                        // 起始时间、结束时间、随机码
                        cacheVo.setStartTime(sessionTo.getStartTime().getTime());
                        cacheVo.setEndTime(sessionTo.getEndTime().getTime());
                        cacheVo.setRandomCode(UUID.randomUUID().toString().replace("-", ""));

                        // 设置虚拟库存 信号量键
                        // 信号量键名使用秒杀商品的随机码
                        RSemaphore semaphore = redissonClient.
                                getSemaphore(SecKillConstant.SEC_KILL_SEMAPHORE + cacheVo.getRandomCode());
                        semaphore.trySetPermits(productTo.getSeckillCount());  // 信号量的值就是当前秒杀商品的备货量
                        // stringRedisTemplate.expire(SecKillConstant.SEC_KILL_SEMAPHORE + cacheVo.getRandomCode(), 30L, TimeUnit.SECONDS);

                        // 键是唯一的场次-商品关联的主键id，而不是skuId
                        productHash.put(productTo.getId().toString(), JSON.toJSONString(cacheVo));
                    }
                }
            }
        }
    }

    /**
     * 前端秒杀商品
     *
     * @param sessionProId 场次-商品关联id。全局唯一
     * @param randomCode   秒杀商品随机码
     * @param skuId        商品skuId，用于和场次-商品关联Id做联合检查
     * @param num          要购买多少件商品
     * @return 创建出的订单号
     */
    @Override
    public R secKillPro(Long sessionProId, String randomCode, Long skuId, Integer num) {
        // 1、登录校验。已经在拦截器中完成

        // 2、合法性校验
        // 从redis中查出场次-商品关联信息，验证skuId
        BoundHashOperations<String, String, String> sessionProHash =
                stringRedisTemplate.boundHashOps(SecKillConstant.SEC_PRODUCT_PREFIX);
        String cacheVoJSON = sessionProHash.get(sessionProId.toString());
        if (StringUtils.isEmpty(cacheVoJSON)) {
            // redis中已经没有这个秒杀商品了
            return R.error("该秒杀活动已结束");
        } else {
            SecKillProductCacheVo cacheVo = JSON.parseObject(cacheVoJSON, SecKillProductCacheVo.class);
            if (!cacheVo.getSecKillProductTo().getSkuId().equals(skuId)) {
                // 合法性校验失败。skuId对不上
                return R.error("合法性校验失败:该商品不在当前秒杀活动中，请严格从秒杀商品详情页进入秒杀活动");
            } else {
                Long startTime = cacheVo.getStartTime();
                Long endTime = cacheVo.getEndTime();
                long curtime = new Date().getTime();
                if (curtime < startTime || curtime > endTime) {
                    // 合法性校验失败。时间不对
                    return R.error("合法性校验失败:该秒杀活动尚未开始，请严格从秒杀商品详情页进入秒杀活动");
                } else if (!cacheVo.getRandomCode().equals(randomCode)) {
                    // 合法性校验失败，随机码不对
                    return R.error("合法性校验失败:您提交了错误的商品秘钥，请严格从秒杀商品详情页进入秒杀活动");
                } else if (cacheVo.getSecKillProductTo().getSeckillLimit() <= num) {
                    // 合法性校验失败，单次购买超过限制
                    return R.error("合法性校验失败:单次购买数量超出限额");
                } else {
                    // 验证当前用户累计购买是否超过限制
                    LoginSuccessTo loginSuccessTo = LoginInterceptor.loginUserThreadLocal.get();
                    String redisKey = SecKillConstant.SEC_PRO_USER + sessionProId.toString() + "_" + loginSuccessTo.getId().toString();
                    String userCount = stringRedisTemplate.opsForValue().get(redisKey);
                    if (!StringUtils.isEmpty(userCount) &&
                            Integer.parseInt(userCount) + num > cacheVo.getSecKillProductTo().getSeckillLimit()) {
                        return R.error("合法性校验失败:您购买此商品的数量已达到限额，请减少购买数量");
                    } else {
                        // 记录下用户最新的累计购买数量。注意此时还不能直接修改redis，因为现在也不能确定用户购买成功
                        userCount = !StringUtils.isEmpty(userCount) ? String.valueOf((Integer.parseInt(userCount) + num)) : num.toString();
                        // 3、 信号量锁
                        RSemaphore semaphore = redissonClient.getSemaphore(SecKillConstant.SEC_KILL_SEMAPHORE + cacheVo.getRandomCode());
                        try {
                            // 判断能否获取到信号量锁。如果能获取到，信号量锁会自动减去num
                            boolean stockLock = semaphore.tryAcquire(num, 10, TimeUnit.MILLISECONDS);
                            if (stockLock) {
                                // 执行到这里，说明所有的前置校验成功，用户秒杀成功
                                // 不再继续向后端放行请求，在这里直接RabbitMQ快速发送消息，然后返回客户端
                                String orderSn = IdWorker.getTimeId();
                                stringRedisTemplate.opsForValue().set(redisKey, userCount); // 设置redis中商品_用户记录为最新的累计购买数量
                                // 构建队列消息并填充这个消息类
                                SecKillOrderTo secKillOrderTo = new SecKillOrderTo();
                                secKillOrderTo.setOrderSn(orderSn);
                                secKillOrderTo.setMemberId(loginSuccessTo.getId());
                                secKillOrderTo.setNum(num);
                                secKillOrderTo.setId(cacheVo.getSecKillProductTo().getId());
                                secKillOrderTo.setPromotionId(cacheVo.getSecKillProductTo().getPromotionId());
                                secKillOrderTo.setPromotionSessionId(cacheVo.getSecKillProductTo().getPromotionSessionId());
                                secKillOrderTo.setSeckillPrice(cacheVo.getSecKillProductTo().getSeckillPrice());
                                rabbitTemplate.convertAndSend("order-event-exchange", "order.seckill.order", secKillOrderTo);
                                return R.ok().put("orderSn", orderSn);
                            } else {
                                return R.error("secKill failed : no more semaphore");
                            }
                        } catch (InterruptedException e) {
                            return R.error("secKill failed : unknown mistake");
                        }
                    }
                }

            }
        }
    }

    /**
     * 前端秒杀。第二种实现，和现有支付框架整合。核心是尽量减少redis和MySQL的访问量
     *
     * @param sessionProId 场次-商品关联id。全局唯一
     * @param randomCode   秒杀商品随机码
     * @param skuId        商品skuId，用于和场次-商品关联Id做联合检查
     * @param num          要购买多少件商品
     * @return R对象。如果成功就把订单号放进去
     */
    @Override
    public R secKillProSec(Long sessionProId, String randomCode, Long skuId, Integer num) {
        // 1、登录校验。已经在拦截器中完成

        // 实现2中，直接检验信号量锁。
        // 注意当秒杀开始时，必然会有一个组件承受全部的秒杀流量。这里就是这里的信号量锁检验
        // 这也是为什么取消了实现1中所有秒杀参数校验，不然这些校验也必将使redis承受全部的秒杀流量
        RSemaphore semaphore = redissonClient.getSemaphore(SecKillConstant.SEC_KILL_SEMAPHORE + randomCode);
        // 检查信号量锁是否大于num
        if (semaphore != null && semaphore.availablePermits() > 0) {
            // 此处信号量锁>0不代表秒杀成功，只代表可以继续前进
            return R.ok();
        } else {
            return R.error("secKill failed : no more semaphore");
        }
    }
}
